package com.test.finkcdc.util;

import com.alibaba.fastjson.JSONObject;
import com.test.finkcdc.config.TargetConfig;
import com.test.finkcdc.entity.JobInfo;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;

public class InsertSink implements Serializable {
    private static final long serialVersionUID = -1099070886257404250L;
    public JobInfo info;

    public InsertSink(JobInfo info) {
        this.info = info;
    }

    public<T> SinkFunction<T> getSinkFunction() {
        return  JdbcSink.sink(info.getInsertSql(), new PostgresBuilder(),
                JdbcExecutionOptions.builder()
                        .withBatchSize(500)
                        .withBatchIntervalMs(200)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(TargetConfig.DRIVER_NAME)
                        .withUrl(TargetConfig.URL)
                        .withUsername(TargetConfig.USER_NAME)
                        .withPassword(TargetConfig.PASSWORD).build());
    }
    public class PostgresBuilder<T> implements JdbcStatementBuilder<T>{

        @Override
        public void accept(PreparedStatement pre, T paramValue) throws SQLException {
            JSONObject jsonObject = JSONObject.parseObject(paramValue + "");
            Object obj = JSONObject.parseObject(jsonObject.getString("jsonBean"), info.getEntity());
            Field[] fields = obj.getClass().getDeclaredFields();
            for(int i=2;i< fields.length;i++){
                Field field=fields[i];
                field.setAccessible(true);
                DateFormatMark dateFormatMark=field.getAnnotation(DateFormatMark.class);
                Object value = null;
                try {
                    value = field.get(obj);
                } catch (IllegalAccessException e) {
                    throw new RuntimeException(e);
                }
                if(dateFormatMark!=null &&value!=null){
                    pre.setObject(i-1,new Timestamp((Long)value-8*60*60*1000));
                }else{
                    pre.setObject(i-1,value);
                }
            }
        }
    }
}
